package com.xlhj.boot.service.impl;

import com.aliyun.odps.*;
import com.aliyun.odps.data.ResultSet;
import com.aliyun.odps.task.SQLTask;
import com.aliyun.odps.tunnel.InstanceTunnel;
import com.xlhj.boot.service.MaxComputeService;
import com.xlhj.boot.util.MaxComputeUtils;
import org.springframework.stereotype.Service;

import com.aliyun.odps.Column;
import com.aliyun.odps.Instance;
import com.aliyun.odps.Odps;
import com.aliyun.odps.OdpsException;
import com.aliyun.odps.PartitionSpec;
import com.aliyun.odps.Table;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.RecordReader;
import com.aliyun.odps.data.ResultSet;
import com.aliyun.odps.task.SQLTask;
import com.aliyun.odps.tunnel.InstanceTunnel;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TunnelException;
import com.aliyun.odps.tunnel.io.TunnelRecordReader;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.common.comm.ResponseMessage;
import com.aliyun.oss.model.AppendObjectRequest;
import com.aliyun.oss.model.AppendObjectResult;
import com.aliyun.oss.model.ObjectMetadata;
import com.xlhj.boot.service.MaxComputeService;
import org.springframework.stereotype.Service;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.List;

/**
 * @author: liucaijing
 * @description: MaxComputeServiceImpl
 * @date: 2021/6/17 16:39
 */
@Service
public class MaxComputeServiceImpl implements MaxComputeService {

    private String accessId = "LTAI5tLwFXY6WAVevkzULV4u";
    private String accessKey = "vT4XCzpkPRXXIxrKzweA4AEHum6QJ7";
    private String endPoint = "http://service.cn-chengdu.maxcompute.aliyun.com/api";
    private String odpsUrl = "http://service.cn-chengdu.maxcompute.aliyun.com/api";
    private String ossEndPoint = "https://oss-cn-chengdu.aliyuncs.com";
    /**
     * 内网时需要设置
     */
    private String tunnelUrl = "http://dt.cn-chengdu.maxcompute.aliyun.com";
    private String project = "default_project_A0C5D51_dev";
    private String table = "sys_user_info";
    private String partition = "";
    private String sql = "SELECT * FROM sys_user_info;";
    private String tempTable = "temp_user_info";



    @Override
    public void getMaxComputeData() {

    }

    @Override
    public void getMoreMaxComputeData() {
        Account account = new AliyunAccount(accessId, accessKey);
        Odps odps = new Odps(account);
        odps.setEndpoint(odpsUrl);
        odps.setDefaultProject(project);
        TableTunnel tunnel = new TableTunnel(odps);
        tunnel.setEndpoint(tunnelUrl);
        PartitionSpec partitionSpec = new PartitionSpec(partition);
        try {
            TableTunnel.DownloadSession downloadSession = tunnel.createDownloadSession(project, table, partitionSpec);
            System.out.println("Session Status is : " + downloadSession.getStatus().toString());
            long count = downloadSession.getRecordCount();
            System.out.println("RecordCount id : " + count);
            RecordReader recordReader = downloadSession.openRecordReader(0, count);
            Record record;
            while (null != (record = recordReader.read())) {
                consumeRecord(record, downloadSession.getSchema());
            }
            recordReader.close();
        } catch (TunnelException e) {
            e.printStackTrace();
        } catch (IOException ei) {
            ei.printStackTrace();
        }
    }

    @Override
    public void getMoreMaxComputeDataForTask() {
       // Odps odps = Odps
    }

    @Override
    public void runSqlTask() {
        Instance instance;
        try {
            instance = SQLTask.run(getOdps(), sql);
            instance.waitForSuccess();
            List<Record> records = SQLTask.getResult(instance);
            for (Record record : records) {
                System.out.println(record.get(0).toString());
                System.out.println(record.getColumns());
                Column[] columns = record.getColumns();
                for (Column column : columns) {
                    String name = column.getName();
                }
                System.out.println(record.clone());
                System.out.println();
            }
        } catch (OdpsException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void dowloadData() {
        Instance instance;
        StringBuilder builder = new StringBuilder("Create Table ").append(tempTable).append(" lifecycle 1 as ").append(sql);
        TableTunnel tunnel = new TableTunnel(getOdps());
        try {
            //instance = SQLTask.run(getOdps(), builder.toString());
            //instance.waitForSuccess();
            TableTunnel.DownloadSession downloadSession = tunnel.createDownloadSession(project, tempTable);
            System.out.println("Session Status is : "+ downloadSession.getStatus().toString());
            long count = downloadSession.getRecordCount();
            System.out.println("RecordCount is: " + count);
            RecordReader recordReader = downloadSession.openRecordReader(0, count);
            Record record;
            while ((record = recordReader.read()) != null) {
                consumeRecord(record, downloadSession.getSchema());
            }
            recordReader.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void getMoreMaxComputeDataForInstance() {
        try {
            Instance instance = SQLTask.run(getOdps(), sql);
            instance.waitForSuccess();
            InstanceTunnel tunnel = new InstanceTunnel(getOdps());
            InstanceTunnel.DownloadSession downloadSession = tunnel.createDownloadSession(getOdps().getDefaultProject(), instance.getId());
            long count = downloadSession.getRecordCount();
            System.out.println(count);
            TunnelRecordReader recordReader = downloadSession.openRecordReader(0, count);
            Record record;
            while (null != (record = recordReader.read())) {
                for (int i = 0; i < downloadSession.getSchema().getColumns().size(); i++) {
                    System.out.println(record.get(i));
                }
            }
            recordReader.close();
        } catch (Exception e) {

        }
    }

    @Override
    public void uploadData() {
        String content1 = "user_id,user_name,phone,user_addr,create_time\n";
        String content2 = "10001,zhangsan,13567898765,重庆,2021-06-18 09:11:21\n10002,lisi,13567898765,重庆,2021-06-18 09:12:21\n";
        try {
            OSS ossClient = new OSSClientBuilder().build(ossEndPoint, accessId, accessKey);
            ObjectMetadata metadata = new ObjectMetadata();

            //上传的内容类型
            metadata.setContentType("text/plain");
            //设置多个参数
            AppendObjectRequest objectRequest = new AppendObjectRequest("data-api-bucket", "cartier/user_info2.csv", new ByteArrayInputStream(content1.getBytes()), metadata);
            //追加数据
            objectRequest.setPosition(0L);
            AppendObjectResult objectResult = ossClient.appendObject(objectRequest);
            System.out.println(objectResult.getObjectCRC());

            objectRequest.setPosition(objectResult.getNextPosition());
            objectRequest.setInputStream(new ByteArrayInputStream(content2.getBytes()));
            objectResult = ossClient.appendObject(objectRequest);
            ResponseMessage response = objectResult.getResponse();
            int code = response.getStatusCode();
            System.out.println(code);
            String uri = response.getUri();
            System.out.println(uri);
            System.out.println(response.toString());
            ossClient.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 只查询5条示例数据即可
     * @return
     * @throws OdpsException
     * @throws IOException
     */
    @Override
    public ResultSet getDataAssetDetail() throws OdpsException, IOException {
        //String sqlStr = "SELECT role_id, role_code, role_name, status, create_time FROM sys_role_info;";
        String sqlStr = "SELECT user_id, user_name, phone, user_addr, create_time FROM sys_user_info;";
        Instance instance = SQLTask.run(getOdps(), sqlStr);
        instance.waitForSuccess();
        //List<Record> recordList = SQLTask.getResult(instance);
        ResultSet resultSet = SQLTask.getResultSet(instance, 5L);
        return resultSet;
    }

    /**
     * 获取表列信息
     * @return
     * @throws OdpsException
     * @throws IOException
     */
    @Override
    public List<Column> getTableColumn() throws OdpsException, IOException {
        Odps odps = getOdps();
        Table table = odps.tables().get("sys_role_info");
        TableSchema schema = table.getSchema();
        List<Column> columns = schema.getColumns();
        return columns;
    }

    private static void consumeRecord(Record record, TableSchema schema) {
        System.out.println(record.getString("user_name"));
    }

    /*private void consumeRecord(Record record, TableSchema schema) {
        for (int i = 0; i < schema.getColumns().size(); i++) {
            Column column = schema.getColumn(i);
            String colValue = null;
            switch (column.getType()) {
                case BIGINT:{
                    Long v = record.getBigint(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case BOOLEAN:{
                    Boolean v = record.getBoolean(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case DATETIME:{
                    Date v = record.getDatetime(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case DOUBLE:{
                    Double v = record.getDouble(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                case STRING:{
                    String v = record.getString(i);
                    colValue = v == null ? null : v.toString();
                    break;
                }
                default:
                    throw new RuntimeException("Unknown column type:" + column.getType());
            }
            System.out.println(colValue == null ? "null" : colValue);
            if (i != schema.getColumns().size()) {
                System.out.println("\t");
            }
            System.out.println();
        }
    }*/

    /**
     * 初始化MaxCompute连接信息
     * @return
     */
    private Odps getOdps() {
        Account account = new AliyunAccount(accessId, accessKey);
        Odps odps = new Odps(account);
        odps.setEndpoint(endPoint);
        odps.setDefaultProject(project);
        return odps;
    }
    @Override
    public ResultSet getDataInfo(String tableName) throws OdpsException, IOException {
        String sqlStr = "SELECT user_id, user_name, phone, user_addr, create_time FROM sys_user_info;";
        Instance instance = SQLTask.run(MaxComputeUtils.getOdps(), sqlStr);
        instance.waitForSuccess();
        ResultSet resultSet = SQLTask.getResultSet(instance, 5L);
        return resultSet;
    }

    @Override
    public List<Column> getTableColumn(String tableName) {
        Odps odps = MaxComputeUtils.getOdps();
        Table table = odps.tables().get(tableName);
        TableSchema schema = table.getSchema();
        List<Column> columns = schema.getColumns();
        return columns;
    }

    @Override
    public InstanceTunnel.DownloadSession getBatchDataInfo(String tableName) throws OdpsException, IOException {
        Odps odps = MaxComputeUtils.getOdps();
        String sqlStr = "SELECT user_id, user_name, phone, user_addr, create_time FROM sys_user_info;";
        Instance instance = SQLTask.run(odps, sqlStr);
        instance.waitForSuccess();
        InstanceTunnel tunnel = new InstanceTunnel(odps);
        InstanceTunnel.DownloadSession downloadSession = tunnel.createDownloadSession(odps.getDefaultProject(), instance.getId());
        return downloadSession;
    }

    @Override
    public void getBatchDataInfoForThread(String tableName) throws OdpsException {
        Odps odps = MaxComputeUtils.getOdps();
        String sqlStr = "SELECT user_id, user_name, phone, user_addr, create_time FROM sys_user_info;";
        Instance instance = SQLTask.run(odps, sqlStr);
        instance.waitForSuccess();
        InstanceTunnel tunnel = new InstanceTunnel(odps);
        InstanceTunnel.DownloadSession downloadSession = tunnel.createDownloadSession(odps.getDefaultProject(), instance.getId());
        return;
    }











}
